package com.nota.flink.cdc.consumer;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.nota.flink.cdc.pojo.DataChangeInfo;
import com.nota.flink.cdc.main.MysqlEventListener;
import com.nota.flink.cdc.pojo.constants.OperatorTypeConstants;
import com.nota.flink.cdc.pojo.constants.RedisStreamConstants;
import com.nota.flink.cdc.utils.SqlParserUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import redis.clients.jedis.UnifiedJedis;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * @author 胡超雄
 * @date 2023/12/26 14:19
 * @desc 消费  RedisStream  里的canal数据
 */
@Component
@Slf4j
public class RedisStreamConsumer {

    @Resource
    private UnifiedJedis unifiedJedis;


    @Resource
    private MysqlEventListener mysqlEventListener;


    /**
     * 消费数据
     *  todo  system  开头的都不要
     *
     * @param dataChangeInfo flink 数据数据实体
     * @return true成功 否则失败
     */
    public Boolean consumeData(DataChangeInfo dataChangeInfo) {
        boolean isSuc = false;
        try {
            //先看是不是sql 语句
            String sql = dataChangeInfo.getDdl();
            if (StrUtil.isEmpty(sql)) {
                Integer operatorType = dataChangeInfo.getOperatorType();
                String tableName = dataChangeInfo.getTableName();
                if ("system_role_auth".equals(tableName)) {
                    return true;
                }
                JSONObject jsonObject = dataChangeInfo.getData();
                String key = RedisStreamConstants.NOTA_SEARCH_PRE + tableName + ":" + jsonObject.getString(mysqlEventListener.TABLE_MAP.get(tableName));
                if (OperatorTypeConstants.DELETE.equals(operatorType)) {
                    // 直接按照id 删除
                    long del = unifiedJedis.del(key);
                    if (del == 1) {
                        isSuc = true;
                    }
                } else if (OperatorTypeConstants.UPDATE.equals(operatorType)
                        || OperatorTypeConstants.ADD.equals(operatorType)
                        || OperatorTypeConstants.INIT.equals(operatorType)) {
                    // 新增和修改都是新增,修改相当于覆盖
                    Map<String, String> hash = new HashMap();
                    hash.put("_language", "chinese");
                    hash.put("agg", "0");
                    for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
                        String curKey = entry.getKey();
                        Object value = entry.getValue();
                        if (value == null) {
                            // 时间设置为null会报错, 所以设置成一个不可能的值,
                            hash.put(curKey, "-1314");
                        } else {
                            String valueStr = value.toString();
                            String validDate = isValidDate(valueStr);
                            if (StrUtil.isNotEmpty(validDate)) {
                                hash.put(curKey, validDate);
                            } else {
                                hash.put(curKey, valueStr);
                            }
                        }
                    }
                    unifiedJedis.hset(key, hash);
                    isSuc = true;
                }
            } else {
                // 处理 sql ,只能处理修改和添加的  ddl语句
                isSuc = SqlParserUtil.parseSql(dataChangeInfo);
            }
        } catch (Exception e) {
            log.error("消费数据异常,当前数据[{}]", JSON.toJSONString(dataChangeInfo), e);
            dataChangeInfo.setErrInfo(e.getMessage());
        }
        return isSuc;
    }

    /**
     * 校验字符串是否是时间格式
     *
     * @param str 需要校验的字符串
     * @return
     */
    public static String isValidDate(String str) {
        // 指定日期格式为四位年/两位月份/两位日期，注意yyyy/MM/dd区分大小写；
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            // 设置lenient为false. 否则SimpleDateFormat会比较宽松地验证日期，比如2007/02/29会被接受，并转换成2007/03/01
            format.setLenient(false);
            Date parse = format.parse(str);
            //
            return parse.toInstant().getEpochSecond() + "";
        } catch (Exception e) {
            // 如果throw java.text.ParseException或者NullPointerException，就说明格式不对
            return null;
        }
    }

    /**
     * 校验字符串是否是时间格式
     *
     * @param object 需要校验的object
     * @return
     */
    public Long isValidDate(Object object) {
        if (ObjectUtil.isEmpty(object)) {
            return null;
        }
        // 指定日期格式为四位年/两位月份/两位日期，注意yyyy/MM/dd区分大小写；
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            // 设置lenient为false. 否则SimpleDateFormat会比较宽松地验证日期，比如2007/02/29会被接受，并转换成2007/03/01
            format.setLenient(false);
            Date parse = format.parse(object.toString());
            //
            return parse.toInstant().getEpochSecond();
        } catch (Exception e) {
            // 如果throw java.text.ParseException或者NullPointerException，就说明格式不对
            return null;
        }
    }


}
